package amqp

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	log "github.com/sirupsen/logrus"
	"github.com/streadway/amqp"
	"io/ioutil"
	"net/http"
	"queue-gateway/config"
	"queue-gateway/databases"
	"queue-gateway/tokenBuket"
	"strconv"
	"time"
)

// TaskResult 任务处理结果
type TaskResult struct {
	Taskid       string              `json:"taskid"`
	URL          string              `json:"url"`
	Status       int                 `json:"status"`
	Header       map[string][]string `json:"header"`
	BodyStr      string              `json:"bodyStr"`
	Body         *json.RawMessage    `json:"body"`
	ConsumedTime string              `json:"consumedTime"`
}

func (t *TaskResult) JSON() ([]byte, error) {
	buffer := &bytes.Buffer{}
	encoder := json.NewEncoder(buffer)
	encoder.SetEscapeHTML(false)
	err := encoder.Encode(t)
	return buffer.Bytes(), err
}

// Consumer 以upstream为单位
func Consumer(ctx context.Context, upstreamName string, upstream config.Upstream) {
	channel := DialChannel()
	defer channel.Close()
	log.Printf("upstream是：%s, 消费者监听队列名：%s", upstreamName, upstream.Queue)
	consume, err := channel.Consume(upstream.Queue, upstreamName, true, false, false, false, nil)
	if err != nil {
		panic(err)
	}
LOOP:
	for {
		select {
		case <-ctx.Done():
			break LOOP
		case msg := <-consume:
			go dispatchTask(ctx, msg, upstream)
		default:
			time.Sleep(time.Millisecond * 10)
		}
	}
}

func postJson(url string, data []byte) (*http.Response, error) {
	var resultErr error
	for i := 0; i < 3; i++ {
		resp, err := http.Post(url, "application/json;", bytes.NewBuffer(data))
		if err == nil {
			//body, _ := ioutil.ReadAll(resp.Body)
			//log.Println(11122222233333, string(body))
			return resp, nil
		} else {
			resultErr = err
		}
	}
	return nil, resultErr
}

func ConsumerInit() {
	for key, upstream := range config.App.Upstreams {
		go Consumer(context.Background(), key, upstream)
	}
}

func returnResponse(ctx context.Context, taskId string, res []byte) error {
	cmd := databases.Rdb.SetEX(ctx, taskId, res, time.Hour)
	if cmd.Err() != nil {
		return cmd.Err()
	}
	return nil
}

func genResponse(ctx context.Context, taskid string, resp *http.Response, resTotal *TaskResult, start time.Time) error {
	if resTotal.Status != 408 && resp != nil {
		body, _ := ioutil.ReadAll(resp.Body)
		bodyJsonRaw := json.RawMessage(body)
		resTotal.Body = &bodyJsonRaw
		resTotal.BodyStr = string(body)
		resTotal.Status = resp.StatusCode
		resTotal.Header = resp.Header
		resTotal.URL = resp.Request.URL.Host + resp.Request.URL.Path
	}

	resTotal.Taskid = taskid
	resTotal.ConsumedTime = strconv.FormatInt(time.Since(start).Milliseconds(), 10) + " Milliseconds"
	dataTotal, _ := resTotal.JSON()
	err := returnResponse(ctx, taskid, dataTotal)
	return err
}

func dispatchTask(ctx context.Context, msg amqp.Delivery, upstream config.Upstream) {
	//callback(msg.Body)
	resTotal := &TaskResult{Status: 200}
	var resp *http.Response
	start := time.Now()
	nowt := time.Now().Sub(msg.Timestamp)
	timeoutLimit := time.Second * time.Duration(upstream.TimeOut)

	// 从MQ中取出的数据超时立即丢弃，不做后续处理
	if nowt >= timeoutLimit {
		log.Debugln(nowt, time.Now().Unix(), msg.Timestamp.Unix())
		resTotal.Status = 408
	} else {
	GETTLOOP:
		for {
			// 超过等待时间还没获取到令牌，则放弃等待令牌，立即返回响应超时
			if time.Now().Sub(start) >= timeoutLimit {
				resTotal.Status = 408
				break GETTLOOP
			}
			hashName, podip, tokenid := tokenBuket.GetToken(ctx, upstream.Services, upstream.Weights)
			if podip != "" {
				if url, ok := msg.Headers["url"]; !ok {
					resTotal.Status = 500
				} else {
					hostUrl := fmt.Sprintf("http://%s%s", podip, url)
					log.Debugf("hostUrl:%s", hostUrl)
					resp, _ = postJson(hostUrl, msg.Body)
					tokenBuket.RevertToken(ctx, hashName, podip, tokenid)
				}
				break GETTLOOP
			}
			time.Sleep(time.Millisecond * 10)
		}

	}

	genResponse(ctx, msg.Headers["taskid"].(string), resp, resTotal, start)
}
